package mq
import(
	"nightowl/logger"
	"time"
    "context"
	"github.com/apache/pulsar-client-go/pulsar"
    // "fmt"
    "os"
)

type LogRecord struct {
    Mod   string    `json:"mod"`
    Grp string `json:"grp"`
    Offset   int    `json:"offset"`
    Msg   string    `json:"msg"`
}
type LogPulsarProducer struct{
    PulsarConnURL string
    client pulsar.Client
    producer pulsar.Producer
    Topic string
}
func (mq *LogPulsarProducer)newPulsarClient() error {
    client, err := pulsar.NewClient(pulsar.ClientOptions{
        URL:               mq.PulsarConnURL,
        OperationTimeout:  30 * time.Second,
        ConnectionTimeout: 30 * time.Second,
    })
    if err != nil {
        logger.Error("newPulsarClient|error: %s", err)
        return err
    }
    mq.client = client
    return nil

}
const (
    NightOwlLogSchema = "{\"type\":\"record\",\"name\":\"nightowl_deploy_log\",\"namespace\":\"test\"," +
        "\"fields\":[{\"name\":\"mod\",\"type\":\"string\"},{\"name\":\"grp\",\"type\":\"string\"},"+
        "{\"name\":\"offset\",\"type\":\"int\"},{\"name\":\"msg\",\"type\":\"string\"}]}"
)

func (mq *LogPulsarProducer)NewPulsarProducer() error {
    

    properties := make(map[string]string)
    properties["pulsar"] = "hello"
    jsonSchemaWithProperties := pulsar.NewJSONSchema(NightOwlLogSchema, properties)
  
    producer, err := mq.client.CreateProducer(pulsar.ProducerOptions{
        Topic:  mq.Topic,
        Schema: jsonSchemaWithProperties,
    })

    if err != nil {
        logger.Error("NewPulsarProducer|error: %s", err)
        return err
    }
    mq.producer = producer
    return nil
}
func (mq LogPulsarProducer)SendPulsarMsg( logRecord *LogRecord ){
    if mq.producer == nil {
        return
    }
    _, err := mq.producer.Send(context.Background(), &pulsar.ProducerMessage{
        Value: logRecord,
    })
    if err != nil {
        logger.Info("SendPulsarMsg|error:", err)
    }else{
        logger.Info("SendPulsarMsg|result:SUCCESS")
    }
    // producer.Close()
}
var LogProducer LogPulsarProducer
func init(){
    pulsarConnURL := os.Getenv("PULSAR_CONN_URL")
    if pulsarConnURL == ""{
        logger.Error("init|Getenv|error:PULSAR_CONN_URL unset")
        return
    }
    logger.Info("init|PULSAR_CONN_URL:", pulsarConnURL)
    LogProducer = LogPulsarProducer{
        Topic: "nightowl-deploy-log",
        PulsarConnURL: pulsarConnURL,

    }
    //创建 pulsar client
    err := LogProducer.newPulsarClient()
    if err != nil {
        logger.Error("init|newPulsarClient|error:PULSAR_CONN_URL unset")
        return
    }
    logger.Info("init|newPulsarClient:OK")
    //创建producer
    err = LogProducer.NewPulsarProducer()
    if err != nil {
        logger.Error("init|NewPulsarProducer|error:PULSAR_CONN_URL unset")
        return
    }    
    logger.Info("init|NewPulsarProducer:OK")
}